package com.sts.sink.impl;

import com.sts.jdbc.JdbcOperation;
import com.sts.jdbc.model.ColumnInfo;
import com.sts.sink.model.JdbcSinkConfig;
import com.sts.source.StsSource;
import com.sts.task.StsTask;

import java.util.List;
import java.util.Map;

/**
 * @author lsl
 * @since 2025年07月04日 11:27
 */
public class OpenGaussSinkImpl extends PostgresqlSinkImpl {
    public OpenGaussSinkImpl(StsSource stsSource, JdbcSinkConfig jdbcSinkConfig, StsTask stsTask, JdbcOperation jdbcOperation) {
        super(stsSource, jdbcSinkConfig, stsTask, jdbcOperation);
    }

    @Override
    public String buildUpsertSql(List<ColumnInfo> columnInfoList, Map<String, ColumnInfo> tableAllPrimaryKeyColumnMap) {
        String identifier = getJdbcSinkConfig().getJdbcConnectionInfo().getIdentifier();

        StringBuilder upsertSql = new StringBuilder("INSERT INTO ");
        upsertSql.append(identifier).append(getJdbcSinkConfig().getSchema()).append(identifier)
                .append(".")
                .append(identifier).append(getJdbcSinkConfig().getTable()).append(identifier)
                .append(" (");

        for (ColumnInfo columnInfo : columnInfoList) {
            upsertSql.append(identifier).append(columnInfo.getColumnName()).append(identifier).append(", ");
        }

        upsertSql.delete(upsertSql.lastIndexOf(","), upsertSql.length());
        upsertSql.append(") VALUES (");

        upsertSql.append("?, ".repeat(columnInfoList.size()));
        upsertSql.delete(upsertSql.lastIndexOf(","), upsertSql.length());

        upsertSql.append(") on duplicate key update ");

        for (ColumnInfo columnInfo : columnInfoList) {
            String columnName = columnInfo.getColumnName();

            // 不是主键的字段需要拼接在 on duplicate key update 后面
            if (!tableAllPrimaryKeyColumnMap.containsKey(columnName.toUpperCase())) {
                upsertSql.append(identifier).append(columnName).append(identifier).append(" = ?, ");
            }
        }

        upsertSql.delete(upsertSql.lastIndexOf(","), upsertSql.length());
        return upsertSql.toString();
    }
}
